package com.zhaotianhao.springboot_demo_fire.config;

import com.alibaba.fastjson.JSON;
import com.zhaotianhao.springboot_demo_fire.entity.Firearms;
import com.zhaotianhao.springboot_demo_fire.entity.Parts;
import com.zhaotianhao.springboot_demo_fire.mapper.FirearmsRepository;
import com.zhaotianhao.springboot_demo_fire.service.IFirearmsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @Author Z-7
 * @Date 2022/3/22
 */
@Slf4j
@Component
@EnableKafka
public class Kafka implements MessageListener<String,String> {
    @Resource
    private RedisTemplate redisTemplate;
    @Autowired
    private IFirearmsService iFirearmsService;
    @Autowired
    private FirearmsRepository firearmsRepository;
    @Override
    @KafkaListener(topics = {"fire"})
    public void onMessage(ConsumerRecord<String, String> data) {
        String key = data.key();
        Firearms firearms = JSON.parseObject(data.value(), Firearms.class);
        // 判断重复消费
        if(!(redisTemplate.opsForValue().setIfAbsent(firearms.getUuid(),"",1, TimeUnit.HOURS))){
            log.info("消息重复进入");
            return;
        }
        log.info("开始正常处理消息");
        if("edit".equals(key)){
            firearmsRepository.save(firearms);
        }else if("delete".equals(key)){
            firearmsRepository.deleteById(firearms.getId());
        }

        // 正常消费 删除备份
        redisTemplate.delete("message:"+firearms.getUuid());

    }
}


